package kafka

import (
	"context"
	"fmt"
	"sync"
	"time"

	"ultone/internal/interfaces"

	"github.com/loveuer/nf/nft/log"
	kfkgo "github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl"
)

var Client *client

type client struct {
	sync.Mutex
	ctx          context.Context
	d            *kfkgo.Dialer
	topic        string
	partition    int
	reconnection bool
	mechanism    sasl.Mechanism
	address      string
	logger       interfaces.Logger
	writer       *kfkgo.Writer
}

func Init(address string, opts ...OptionFn) error {
	c, err := New(address, opts...)
	if err != nil {
		return err
	}

	Client = c

	return nil
}

func New(address string, opts ...OptionFn) (*client, error) {
	c := &client{}

	if address == "" {
		return nil, fmt.Errorf("address required")
	}

	for _, fn := range opts {
		fn(c)
	}

	c.address = address

	if c.ctx == nil {
		c.ctx = context.Background()
	}

	if c.logger == nil {
		c.logger = log.New()
	}

	dia := &kfkgo.Dialer{
		Timeout: 30 * time.Second,
	}

	if c.mechanism != nil {
		dia.SASLMechanism = c.mechanism
	}

	c.d = dia

	return c, nil
}
